package com.wudl.flink.sql;

import com.wudl.flink.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.table.types.DataType;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author ：wudl
 * @date ：Created in 2021/11/21 21:59
 * @description：
 * @modified By：
 * @version: 1.0
 */

public class Flink_TableApi_ToFileSystem {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        DataStreamSource<WaterSensor> streamSource = env.fromElements(new WaterSensor("sensor_1", 1000L, 10),
                new WaterSensor("sensor_1", 2000L, 20),
                new WaterSensor("sensor_2", 3000L, 30),
                new WaterSensor("sensor_1", 4000L, 40),
                new WaterSensor("sensor_2", 5000L, 50),
                new WaterSensor("sensor_1", 6000L, 60)
        );
        //1.将流转化为动态表
        Table table = tableEnv.fromDataStream(streamSource);
        //2. 对动态表进行查询
        Table resultTable = table.where($("id").isEqual("sensor_1")).select($("id"), $("ts"), $("vc"));

        //resultTable.execute().print();
         // 3.创建输出表
        Schema schema = new Schema().field("id", DataTypes.STRING())
                .field("ts",DataTypes.BIGINT())
                .field("vc",DataTypes.INT());

        tableEnv.connect(new FileSystem().path("input/result.txt"))
                .withFormat(new Csv().fieldDelimiter('|'))
                .withSchema(schema)
                .createTemporaryTable("sensor");

        resultTable.executeInsert("sensor");





        System.out.println("------------------------------------------------------------------------------------------------------");
        // 把动态表转化为流
//        DataStream<Row> rowDataStream = tableEnv.toAppendStream(resultTable, Row.class);
//        rowDataStream.print();
//        env.execute();

    }
}
